package za.org.jcicada.comms.impl.udp;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import za.org.jcicada.comms.IMessageListener;
import za.org.jcicada.comms.IMessageSerializerStrategy;
import za.org.jcicada.comms.impl.ConnectionBase;
import za.org.jcicada.comms.impl.JavaMessageSerializer;
import za.org.jcicada.comms.impl.MessageExecutor;
import za.org.jcicada.comms.impl.tcp.TcpSelector;
import za.org.jcicada.comms.impl.tcp.TcpSelector.ICallback;
import za.org.jcicada.comms.messages.CommsMessage;

public class UdpConnection extends ConnectionBase implements ICallback {

	private static Log logger = LogFactory.getLog(UdpConnection.class);
	private DatagramChannel channel;
	private IMessageSerializerStrategy serializer = new JavaMessageSerializer();
	
	public UdpConnection(DatagramChannel channel) {
		this.channel = channel;
	}
	
	@Override
	public void initialize(IMessageListener listener) throws IOException {
		super.initialize(listener);
		TcpSelector.getInstance().addReader(channel, this);
	}
	
	@Override
	public void close() throws IOException {
		channel.close();
		try {
			getListener().onClose();
		} catch (Exception e) {
			throw new IOException(e);
		}
	}

	@Override
	public boolean isClosed() {
		return !channel.isOpen();
	}

	@Override
	public void sendMessage(CommsMessage message) throws IOException {
		ByteBuffer sendBuffer = ByteBuffer.wrap(serializer.toBytes(message));
		TcpSelector.getInstance().send(channel, sendBuffer);
	}
	
	@Override
	public int read(ByteBuffer dst) throws IOException {
		dst.clear();
		int numRead = channel.read(dst);
		receiveBuffer(dst);
		return numRead;
	}

	@Override
	public int write(ByteBuffer src) throws IOException {
		return channel.write(src);
	}
	
	protected void receiveBuffer(ByteBuffer buffer) {
		try {
			CommsMessage message = serializer.fromBytes(buffer.array());
			MessageExecutor.getInstance().submit(getListener(), message);
		} catch (IOException e) {
			logger.error("Error deserializing received message", e);
		}
	}
	
	@Override
	public String toString() {
		return super.toString() + ":" + channel.socket().getLocalSocketAddress();
	}
	
	protected DatagramChannel getDatagramChannel() {
		return channel;
	}
	
}
